package com.chen.spikemall.redis;

import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;

import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class BlockingRedisLock implements Lock {

    private Object notifyLock;

    private String uuid = UUID.randomUUID().toString();

    private static final String REDIS_LOCK_PREX = "redis_lock_";

    private String lockKey;

    private StringRedisTemplate redisTemplate;

    private static final Long TTL = 10000L;

    private static final String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return -1 end";

    private RedisLockNotifyProxy redisLockNotifyProxy;

    private String threadName;

    public void setRedisTemplate(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void setRedisLockNotifyProxy(RedisLockNotifyProxy redisLockNotifyProxy) {
        this.redisLockNotifyProxy = redisLockNotifyProxy;
    }

    public BlockingRedisLock(String key) {
        this.lockKey = REDIS_LOCK_PREX + key;
    }

    public void setNotifyLock(Object notifyLock) {
        this.notifyLock = notifyLock;
    }

    public String getLockKey() {
        return lockKey;
    }

    @Override
    public void lock() {
        final String lockKeyTemp = lockKey;
        threadName = Thread.currentThread().getName();

        while (true) {
            Boolean ret = redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection) throws DataAccessException {

                    Boolean result = connection.set(lockKeyTemp.getBytes(), threadName.getBytes(), Expiration.milliseconds(TTL), RedisStringCommands.SetOption.SET_IF_ABSENT);
                    return result;
                }
            });

            if (ret != null && ret) {
                return;
            } else {

                if (notifyLock != null) {
                    synchronized (notifyLock) {
                        try {
                            notifyLock.wait(TTL/2);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                } else {
                    try {
                        Thread.sleep(TTL/2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean tryLock() {
        final String lockKeyTemp = lockKey;
        threadName = Thread.currentThread().getName();

        Boolean ret = redisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(RedisConnection connection) throws DataAccessException {

                Boolean result = connection.set(lockKeyTemp.getBytes(), threadName.getBytes(), Expiration.milliseconds(TTL), RedisStringCommands.SetOption.SET_IF_ABSENT);
                return result;
            }
        });

        return ret != null ? ret : false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        throw new UnsupportedOperationException();
    }

    @Override
    public void unlock() {
        final String lockKeyTemp = lockKey;

        RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
        connection.eval(luaScript.getBytes(), ReturnType.INTEGER, 1, (lockKeyTemp).getBytes(), threadName.getBytes());

        redisLockNotifyProxy.lockPublish(this);
    }

    @Override
    public Condition newCondition() {
        throw new UnsupportedOperationException();
    }


    @Override
    public String toString() {
        return "BlockingRedisLock{" +
                "uuid='" + uuid + '\'' +
                ", lockKey='" + lockKey + '\'' +
                ", threadId='" + threadName + '\'' +
                '}';
    }
}
